package db.app

import com.cw.realtime.common.base.FlinkRunner.{RunnerConfig, SqlRunnerConfig, createTopicDbTable, runSql}
import com.cw.realtime.common.constant.Constant._
import com.cw.realtime.common.util.SqlUtil
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment

object DwdTradeCartAdd {

  def main(args: Array[String]): Unit = {

    implicit val config = SqlRunnerConfig("dwd_trade_cart_add", 10013)
    runSql { (env, tableEnv) =>

      createTopicDbTable(tableEnv)

      // 过滤出购物车加购表
      val cartAdd = filterCartAdd(tableEnv)

      createKafkaSink(tableEnv)

      cartAdd.insertInto(TOPIC_DWD_TRADE_CART_ADD).execute()


    }


  }

  def filterCartAdd(tableEnv: StreamTableEnvironment) = {
    tableEnv.sqlQuery(
      """
select
data['id']          id            ,
data['user_id']     user_id         ,
data['sku_id']      sku_id          ,
data['cart_price']  cart_price        ,
if(`type`='insert', `data`['sku_num'], cast(cast(`data`['sku_num'] as bigint) - cast(`old`['sku_num'] as bigint)as string)) `sku_num`           ,
data['sku_name']      sku_name         ,
data['is_checked']    is_checked        ,
data['create_time']   create_time       ,
data['operate_time']  operate_time       ,
data['is_ordered']    is_ordered        ,
data['order_time']    order_time      ,
ts

from topic_db
where
`database` = 'gmall' and
`table` = 'cart_info' and
 (`type` = 'insert' or ( `type` = 'update' and `data`['sku_num']  is  not null
 and  cast(`data`['sku_num'] as bigint) > cast(`old`['sku_num'] as bigint)))

        """
    )
  }


  def createKafkaSink(tableEnv: StreamTableEnvironment) = {

    tableEnv.executeSql(
      s"""
    create table $TOPIC_DWD_TRADE_CART_ADD (
     id          string,
     user_id     string,
     sku_id      string,
     cart_price  string,
     sku_num     string,
     sku_name     string,
     is_checked   string,
     create_time  string,
     operate_time string,
     is_ordered   string,
     order_time   string,
     ts           bigint
       )
     """ + SqlUtil.getKafkaSinkSql(TOPIC_DWD_TRADE_CART_ADD))

  }


}
